package drds.data_propagate.parse;

import com.google.protobuf.ByteString;
import drds.data_propagate.binlog.BinLogEvent;
import drds.data_propagate.binlog.event.Header;
import drds.data_propagate.binlog.event.*;
import drds.data_propagate.binlog.event.TableMapEvent.ColumnInfo;
import drds.data_propagate.binlog.event.mariadb.AnnotateRowsEvent;
import drds.data_propagate.binlog.event.rows_event.DeleteRowsEvent;
import drds.data_propagate.binlog.event.rows_event.RowsEvent;
import drds.data_propagate.binlog.event.rows_event.UpdateRowsEvent;
import drds.data_propagate.binlog.event.rows_event.WriteRowsEvent;
import drds.data_propagate.binlog.exception.TableIdNotFoundException;
import drds.data_propagate.common.AbstractLifeCycle;
import drds.data_propagate.filter.aviater.AviaterRegexFilter;
import drds.data_propagate.parse.exception.ParseException;
import drds.data_propagate.parse.table_meta_data.ColumnMetaData;
import drds.data_propagate.parse.table_meta_data.TableMetaCache;
import drds.data_propagate.parse.table_meta_data.TableMetaData;
import drds.data_propagate.parse.table_meta_data.ddl.DdlResult;
import drds.data_propagate.parse.table_meta_data.ddl.DruidDdlParser;
import drds.data_propagate.parse.table_meta_data.ddl.SimpleDdlParser;
import drds.propagate.protocol.CanalEntry.*;
import drds.propagate.protocol.Entry;
import drds.propagate.protocol.position.EntryPosition;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.charset.Charset;
import java.sql.Types;
import java.util.Arrays;
import java.util.BitSet;
import java.util.List;

/**
 * 基于{@linkplain BinLogEvent}转化为Entry对象的处理
 *
 * @author jianghang 2013-1-17 下午02:41:14
 * @version 1.0.0
 */
public class LogEventConvert extends AbstractLifeCycle implements BinlogParser<BinLogEvent> {

    public static final String XA_XID = "XA_XID";
    public static final String XA_TYPE = "XA_TYPE";
    public static final String XA_START = "XA START";
    public static final String XA_END = "XA END";
    public static final String XA_COMMIT = "XA COMMIT";
    public static final String XA_ROLLBACK = "XA ROLLBACK";
    public static final String ISO_8859_1 = "ISO-8859-1";
    public static final String UTF_8 = "UTF-8";
    public static final int TINYINT_MAX_VALUE = 256;
    public static final int SMALLINT_MAX_VALUE = 65536;
    public static final int MEDIUMINT_MAX_VALUE = 16777216;
    public static final long INTEGER_MAX_VALUE = 4294967296L;
    public static final BigInteger BIGINT_MAX_VALUE = new BigInteger("18446744073709551616");
    public static final int version = 1;
    public static final String BEGIN = "BEGIN";
    public static final String COMMIT = "COMMIT";
    public static final Logger logger = LoggerFactory.getLogger(LogEventConvert.class);

    private volatile AviaterRegexFilter nameFilter;                                                          // 运行时引用可能会有变化，比如规则发生变化时
    private volatile AviaterRegexFilter nameBlackFilter;

    private TableMetaCache tableMetaCache;
    private Charset charset = Charset.defaultCharset();
    private boolean filterQueryDcl = false;
    private boolean filterQueryDml = false;
    private boolean filterQueryDdl = false;
    // 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92
    private boolean filterTableError = false;
    // 新增rows过滤，用于仅订阅除rows以外的数据
    private boolean filterRows = false;
    private boolean useDruidDdlFilter = true;

    public LogEventConvert() {

    }

    public static TransactionBegin createTransactionBegin(long threadId) {
        TransactionBegin.Builder beginBuilder = TransactionBegin.newBuilder();
        beginBuilder.setThreadId(threadId);
        return beginBuilder.build();
    }

    public static TransactionEnd createTransactionEnd(long transactionId) {
        TransactionEnd.Builder endBuilder = TransactionEnd.newBuilder();
        endBuilder.setTransactionId(String.valueOf(transactionId));
        return endBuilder.build();
    }

    public static Pair createSpecialPair(String key, String value) {
        Pair.Builder pairBuilder = Pair.newBuilder();
        pairBuilder.setKey(key);
        pairBuilder.setValue(value);
        return pairBuilder.build();
    }

    public static Entry createEntry(drds.propagate.protocol.CanalEntry.Header header, EntryType entryType, ByteString storeValue) {
        Entry.Builder entryBuilder = Entry.newBuilder();
        entryBuilder.setHeader(header);
        entryBuilder.setEntryType(entryType);
        entryBuilder.setStoreValue(storeValue);
        return entryBuilder.build();
    }

    @Override
    public Entry parse(BinLogEvent binLogEvent, boolean isSeek) throws ParseException {
        if (binLogEvent == null || binLogEvent instanceof UnknownEvent) {
            return null;
        }

        int eventType = binLogEvent.getHeader().getEventType();
        switch (eventType) {
            case BinLogEvent.QUERY_EVENT:
                return parseQueryEvent((QueryEvent) binLogEvent, isSeek);
            case BinLogEvent.XID_EVENT:
                return parseXidEvent((XidEvent) binLogEvent);
            case BinLogEvent.TABLE_MAP_EVENT:
                break;
            case BinLogEvent.WRITE_ROWS_EVENT_V1:
            case BinLogEvent.WRITE_ROWS_EVENT:
                return parseRowsEvent((WriteRowsEvent) binLogEvent);
            case BinLogEvent.UPDATE_ROWS_EVENT_V1:
            case BinLogEvent.PARTIAL_UPDATE_ROWS_EVENT:
            case BinLogEvent.UPDATE_ROWS_EVENT:
                return parseRowsEvent((UpdateRowsEvent) binLogEvent);
            case BinLogEvent.DELETE_ROWS_EVENT_V1:
            case BinLogEvent.DELETE_ROWS_EVENT:
                return parseRowsEvent((DeleteRowsEvent) binLogEvent);
            case BinLogEvent.ROWS_QUERY_LOG_EVENT:
                return parseRowsQueryEvent((RowsQueryEvent) binLogEvent);
            case BinLogEvent.ANNOTATE_ROWS_EVENT:
                return parseAnnotateRowsEvent((AnnotateRowsEvent) binLogEvent);
            case BinLogEvent.USER_VAR_EVENT:
                return parseUserVarLogEvent((UserVarEvent) binLogEvent);
            case BinLogEvent.INTVAR_EVENT:
                return parseIntrvarLogEvent((IntvarEvent) binLogEvent);
            case BinLogEvent.RAND_EVENT:
                return parseRandLogEvent((RandEvent) binLogEvent);
            case BinLogEvent.GTID_LOG_EVENT:
                return parseGTIDLogEvent((GtidEvent) binLogEvent);
            case BinLogEvent.HEARTBEAT_LOG_EVENT:
                return parseHeartbeatLogEvent((HeartbeatEvent) binLogEvent);
            default:
                break;
        }

        return null;
    }

    public void reset() {
        // do nothing
        if (tableMetaCache != null) {
            tableMetaCache.clearTableMeta();
        }
    }

    private Entry parseHeartbeatLogEvent(HeartbeatEvent logEvent) {
        drds.propagate.protocol.CanalEntry.Header.Builder headerBuilder = drds.propagate.protocol.CanalEntry.Header.newBuilder();
        headerBuilder.setEventType(EventType.MHEARTBEAT);
        Entry.Builder entryBuilder = Entry.newBuilder();
        entryBuilder.setHeader(headerBuilder.build());
        entryBuilder.setEntryType(EntryType.heartbeat);
        return entryBuilder.build();
    }

    private Entry parseGTIDLogEvent(GtidEvent logEvent) {
        Header logHeader = logEvent.getHeader();
        Pair.Builder builder = Pair.newBuilder();
        builder.setKey("gtid");
        builder.setValue(logEvent.getGtidStr());

        if (logEvent.getLastCommitted() != null) {
            builder.setKey("lastCommitted");
            builder.setValue(String.valueOf(logEvent.getLastCommitted()));
            builder.setKey("sequenceNumber");
            builder.setValue(String.valueOf(logEvent.getSequenceNumber()));
        }

        drds.propagate.protocol.CanalEntry.Header header = createHeader(logHeader, "", "", EventType.GTID);
        return createEntry(header, EntryType.gtid_log, builder.build().toByteString());
    }

    private Entry parseQueryEvent(QueryEvent event, boolean isSeek) {
        String queryString = event.getQuery();
        if (StringUtils.startsWithIgnoreCase(queryString, XA_START)) {
            // xa start use TransactionBegin
            TransactionBegin.Builder beginBuilder = TransactionBegin.newBuilder();
            beginBuilder.setThreadId(event.getSessionId());
            beginBuilder.addProps(createSpecialPair(XA_TYPE, XA_START));
            beginBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_START)));
            TransactionBegin transactionBegin = beginBuilder.build();
            drds.propagate.protocol.CanalEntry.Header header = createHeader(event.getHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_begin, transactionBegin.toByteString());
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_END)) {
            // xa start use TransactionEnd
            TransactionEnd.Builder endBuilder = TransactionEnd.newBuilder();
            endBuilder.setTransactionId(String.valueOf(0L));
            endBuilder.addProps(createSpecialPair(XA_TYPE, XA_END));
            endBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_END)));
            TransactionEnd transactionEnd = endBuilder.build();
            drds.propagate.protocol.CanalEntry.Header header = createHeader(event.getHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_end, transactionEnd.toByteString());
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_COMMIT)) {
            // xa commit
            drds.propagate.protocol.CanalEntry.Header header = createHeader(event.getHeader(), "", "", EventType.XACOMMIT);
            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            rowChangeBuider.setSql(queryString);
            rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_COMMIT));
            rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_COMMIT)));
            rowChangeBuider.setEventType(EventType.XACOMMIT);
            return createEntry(header, EntryType.row_data, rowChangeBuider.build().toByteString());
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_ROLLBACK)) {
            // xa rollback
            drds.propagate.protocol.CanalEntry.Header header = createHeader(event.getHeader(), "", "", EventType.XAROLLBACK);
            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            rowChangeBuider.setSql(queryString);
            rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_ROLLBACK));
            rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_ROLLBACK)));
            rowChangeBuider.setEventType(EventType.XAROLLBACK);
            return createEntry(header, EntryType.row_data, rowChangeBuider.build().toByteString());
        } else if (StringUtils.endsWithIgnoreCase(queryString, BEGIN)) {
            TransactionBegin transactionBegin = createTransactionBegin(event.getSessionId());
            drds.propagate.protocol.CanalEntry.Header header = createHeader(event.getHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_begin, transactionBegin.toByteString());
        } else if (StringUtils.endsWithIgnoreCase(queryString, COMMIT)) {
            TransactionEnd transactionEnd = createTransactionEnd(0L); // MyISAM可能不会有xid事件
            drds.propagate.protocol.CanalEntry.Header header = createHeader(event.getHeader(), "", "", null);
            return createEntry(header, EntryType.transaction_end, transactionEnd.toByteString());
        } else {
            boolean notFilter = false;
            EventType type = EventType.QUERY;
            String tableName = null;
            String schemaName = null;
            if (useDruidDdlFilter) {
                List<DdlResult> ddlResultList = DruidDdlParser.parse(queryString, event.getDbName());
                for (DdlResult result : ddlResultList) {
                    if (!processFilter(queryString, result)) {
                        // 只要有一个数据不进行过滤
                        notFilter = true;
                    }
                }
                if (ddlResultList.size() > 0) {
                    // 如果针对多行的DDL,只能取第一条
                    type = ddlResultList.get(0).getEventType();
                    schemaName = ddlResultList.get(0).getSchemaName();
                    tableName = ddlResultList.get(0).getTableName();
                }
            } else {
                DdlResult ddlResult = SimpleDdlParser.parse(queryString, event.getDbName());
                if (!processFilter(queryString, ddlResult)) {
                    notFilter = true;
                }

                type = ddlResult.getEventType();
                schemaName = ddlResult.getSchemaName();
                tableName = ddlResult.getTableName();
            }

            if (!notFilter) {
                // 如果是过滤的数据就不处理了
                return null;
            }

            boolean isDml = (type == EventType.INSERT || type == EventType.UPDATE || type == EventType.DELETE);

            if (!isSeek && !isDml) {
                // 使用新的表结构元数据管理方式
                EntryPosition position = createPosition(event.getHeader());
                tableMetaCache.apply(position, event.getDbName(), queryString, null);
            }

            drds.propagate.protocol.CanalEntry.Header header = createHeader(event.getHeader(), schemaName, tableName, type);
            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            if (type != EventType.QUERY && !isDml) {
                rowChangeBuider.setIsDdl(true);
            }
            rowChangeBuider.setSql(queryString);
            if (StringUtils.isNotEmpty(event.getDbName())) {// 可能为空
                rowChangeBuider.setDdlSchemaName(event.getDbName());
            }
            rowChangeBuider.setEventType(type);
            return createEntry(header, EntryType.row_data, rowChangeBuider.build().toByteString());
        }
    }

    private String getXaXid(String queryString, String type) {
        return StringUtils.substringAfter(queryString, type);
    }

    private boolean processFilter(String queryString, DdlResult result) {
        String schemaName = result.getSchemaName();
        String tableName = result.getTableName();
        // fixed issue https://github.com/alibaba/canal/issues/58
        // 更新下table meta cache
        if (tableMetaCache != null
                && (result.getEventType() == EventType.ALTER || result.getEventType() == EventType.ERASE || result.getEventType() == EventType.RENAME)) {
            // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
            for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) {
                String schemaName0 = renameResult.getSchemaName();
                String tableName0 = renameResult.getTableName();
                if (StringUtils.isNotEmpty(tableName0)) {
                    // 如果解析到了正确的表信息，则根据全名进行清除
                    tableMetaCache.clearTableMeta(schemaName0, tableName0);
                } else {
                    // 如果无法解析正确的表信息，则根据schema进行清除
                    tableMetaCache.clearTableMetaWithSchemaName(schemaName0);
                }
            }
        }

        // fixed issue https://github.com/alibaba/canal/issues/58
        if (result.getEventType() == EventType.ALTER || result.getEventType() == EventType.ERASE
                || result.getEventType() == EventType.CREATE || result.getEventType() == EventType.TRUNCATE
                || result.getEventType() == EventType.RENAME || result.getEventType() == EventType.CINDEX
                || result.getEventType() == EventType.DINDEX) { // 针对DDL类型

            if (filterQueryDdl) {
                return true;
            }

            if (StringUtils.isEmpty(tableName)
                    || (result.getEventType() == EventType.RENAME && StringUtils.isEmpty(result.getOriTableName()))) {
                // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析
                throw new ParseException("SimpleDdlParser process query failed. pls submit issue with this queryString: "
                        + queryString + " , and DdlResult: " + result.toString());
                // return null;
            } else {
                // check name filter
                String name = schemaName + "." + tableName;
                if (nameFilter != null && !nameFilter.filter(name)) {
                    if (result.getEventType() == EventType.RENAME) {
                        // rename校验只要源和目标满足一个就进行操作
                        if (nameFilter != null
                                && !nameFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) {
                            return true;
                        }
                    } else {
                        // 其他情况返回null
                        return true;
                    }
                }

                if (nameBlackFilter != null && nameBlackFilter.filter(name)) {
                    if (result.getEventType() == EventType.RENAME) {
                        // rename校验只要源和目标满足一个就进行操作
                        if (nameBlackFilter != null
                                && nameBlackFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) {
                            return true;
                        }
                    } else {
                        // 其他情况返回null
                        return true;
                    }
                }
            }
        } else if (result.getEventType() == EventType.INSERT || result.getEventType() == EventType.UPDATE
                || result.getEventType() == EventType.DELETE) {
            // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
            if (filterQueryDml) {
                return true;
            }
        } else if (filterQueryDcl) {
            return true;
        }

        return false;
    }

    private Entry parseRowsQueryEvent(RowsQueryEvent event) {
        if (filterQueryDml) {
            return null;
        }
        // mysql5.6支持，需要设置binlog-rows-query-log-events=1，可详细打印原始DML语句
        String queryString = null;
        try {
            queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset.name());
            String tableName = null;
            if (useDruidDdlFilter) {
                List<DdlResult> results = DruidDdlParser.parse(queryString, null);
                if (results.size() > 0) {
                    tableName = results.get(0).getTableName();
                }
            }

            return buildQueryEntry(queryString, event.getHeader(), tableName);
        } catch (UnsupportedEncodingException e) {
            throw new ParseException(e);
        }
    }

    private Entry parseAnnotateRowsEvent(AnnotateRowsEvent annotateRowsEvent) {
        if (filterQueryDml) {
            return null;
        }
        // mariaDb支持，需要设置binlog_annotate_row_events=true，可详细打印原始DML语句
        String queryString = null;
        try {
            queryString = new String(annotateRowsEvent.getRowsQuery().getBytes(ISO_8859_1), charset.name());
            return buildQueryEntry(annotateRowsEvent.getHeader(), queryString);
        } catch (UnsupportedEncodingException e) {
            throw new ParseException(e);
        }
    }

    private Entry parseUserVarLogEvent(UserVarEvent userVarEvent) {
        if (filterQueryDml) {
            return null;
        }

        return buildQueryEntry(userVarEvent.getHeader(), userVarEvent.getQuery());
    }

    private Entry parseIntrvarLogEvent(IntvarEvent intvarEvent) {
        if (filterQueryDml) {
            return null;
        }

        return buildQueryEntry(intvarEvent.getHeader(), intvarEvent.getQuery());
    }

    private Entry parseRandLogEvent(RandEvent randEvent) {
        if (filterQueryDml) {
            return null;
        }

        return buildQueryEntry(randEvent.getHeader(), randEvent.getQuery());
    }

    private Entry parseXidEvent(XidEvent xidEvent) {
        TransactionEnd transactionEnd = createTransactionEnd(xidEvent.getXid());
        drds.propagate.protocol.CanalEntry.Header header = createHeader(xidEvent.getHeader(), "", "", null);
        return createEntry(header, EntryType.transaction_end, transactionEnd.toByteString());
    }

    public TableMetaData parseRowsEventForTableMeta(RowsEvent rowsEvent) {
        TableMapEvent rowsEventTable = rowsEvent.getTableMapEvent();
        if (rowsEventTable == null) {
            // tableId对应的记录不存在
            throw new TableIdNotFoundException("not found tableId:" + rowsEvent.getTableId());
        }

        boolean isHeartBeat = isAliSQLHeartBeat(rowsEventTable.getDbName(), rowsEventTable.getTableName());
        boolean isRDSHeartBeat = tableMetaCache.isOnRDS() && isRDSHeartBeat(rowsEventTable.getDbName(), rowsEventTable.getTableName());

        String fullname = rowsEventTable.getDbName() + "." + rowsEventTable.getTableName();
        // check name filter
        if (nameFilter != null && !nameFilter.filter(fullname)) {
            return null;
        }
        if (nameBlackFilter != null && nameBlackFilter.filter(fullname)) {
            return null;
        }

        // if (isHeartBeat || isRDSHeartBeat) {
        // // 忽略rds模式的mysql.ha_health_check心跳数据
        // return null;
        // }
        TableMetaData tableMetaData = null;
        if (isRDSHeartBeat) {
            // 处理rds模式的mysql.ha_health_check心跳数据
            // 主要RDS的心跳表基本无权限,需要mock一个tableMeta
            ColumnMetaData id = new ColumnMetaData("id", "bigint(20)", true, false, "0");
            ColumnMetaData type = new ColumnMetaData("eventType", "char(1)", false, true, "0");
            tableMetaData = new TableMetaData(rowsEventTable.getDbName(), rowsEventTable.getTableName(), Arrays.asList(id, type));
        } else if (isHeartBeat) {
            // 处理alisql模式的test.heartbeat心跳数据
            // 心跳表基本无权限,需要mock一个tableMeta
            ColumnMetaData idMeta = new ColumnMetaData("id", "smallint(6)", false, true, null);
            ColumnMetaData typeMeta = new ColumnMetaData("ts", "int(11)", true, false, null);
            tableMetaData = new TableMetaData(rowsEventTable.getDbName(), rowsEventTable.getTableName(), Arrays.asList(idMeta, typeMeta));
        }

        EntryPosition entryPosition = createPosition(rowsEvent.getHeader());
        if (tableMetaCache != null && tableMetaData == null) {// 入错存在table meta
            tableMetaData = getTableMeta(rowsEventTable.getDbName(), rowsEventTable.getTableName(), true, entryPosition);
            if (tableMetaData == null) {
                if (!filterTableError) {
                    throw new ParseException("not found [" + fullname + "] in db , pls check!");
                }
            }
        }

        return tableMetaData;
    }

    public Entry parseRowsEvent(RowsEvent rowsEvent) {
        return parseRowsEvent(rowsEvent, null);
    }

    public Entry parseRowsEvent(RowsEvent rowsEvent, TableMetaData tableMetaData) {
        if (filterRows) {
            return null;
        }
        try {
            if (tableMetaData == null) { // 如果没有外部指定
                tableMetaData = parseRowsEventForTableMeta(rowsEvent);
            }

            if (tableMetaData == null) {
                // 拿不到表结构,执行忽略
                return null;
            }

            EventType eventType = null;
            int type = rowsEvent.getHeader().getEventType();
            if (BinLogEvent.WRITE_ROWS_EVENT_V1 == type || BinLogEvent.WRITE_ROWS_EVENT == type) {
                eventType = EventType.INSERT;
            } else if (BinLogEvent.UPDATE_ROWS_EVENT_V1 == type || BinLogEvent.UPDATE_ROWS_EVENT == type
                    || BinLogEvent.PARTIAL_UPDATE_ROWS_EVENT == type) {
                eventType = EventType.UPDATE;
            } else if (BinLogEvent.DELETE_ROWS_EVENT_V1 == type || BinLogEvent.DELETE_ROWS_EVENT == type) {
                eventType = EventType.DELETE;
            } else {
                throw new ParseException("unsupport event eventType :" + rowsEvent.getHeader().getEventType());
            }

            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            rowChangeBuider.setTableId(rowsEvent.getTableId());
            rowChangeBuider.setIsDdl(false);

            rowChangeBuider.setEventType(eventType);
            RowsLogBuffer buffer = rowsEvent.getRowsBuf(charset.name());
            BitSet columns = rowsEvent.getColumns();
            BitSet changeColumns = rowsEvent.getChangeColumns();

            boolean tableError = false;
            int rowsCount = 0;
            while (buffer.nextOneRow(columns, false)) {
                // 处理row记录
                RowData.Builder rowDataBuilder = RowData.newBuilder();
                if (EventType.INSERT == eventType) {
                    // insert的记录放在before字段中
                    tableError |= parseOneRow(rowDataBuilder, rowsEvent, buffer, columns, true, tableMetaData);
                } else if (EventType.DELETE == eventType) {
                    // delete的记录放在before字段中
                    tableError |= parseOneRow(rowDataBuilder, rowsEvent, buffer, columns, false, tableMetaData);
                } else {
                    // update需要处理before/after
                    tableError |= parseOneRow(rowDataBuilder, rowsEvent, buffer, columns, false, tableMetaData);
                    if (!buffer.nextOneRow(changeColumns, true)) {
                        rowChangeBuider.addRowDatas(rowDataBuilder.build());
                        break;
                    }

                    tableError |= parseOneRow(rowDataBuilder, rowsEvent, buffer, changeColumns, true, tableMetaData);
                }

                rowsCount++;
                rowChangeBuider.addRowDatas(rowDataBuilder.build());
            }
            TableMapEvent table = rowsEvent.getTableMapEvent();
            drds.propagate.protocol.CanalEntry.Header header = createHeader(rowsEvent.getHeader(),
                    table.getDbName(),
                    table.getTableName(),
                    eventType,
                    rowsCount);

            RowChange rowChange = rowChangeBuider.build();
            if (tableError) {
                Entry entry = createEntry(header, EntryType.row_data, ByteString.EMPTY);
                logger.warn("table parser error : {}storeValue: {}", entry.toString(), rowChange.toString());
                return null;
            } else {
                Entry entry = createEntry(header, EntryType.row_data, rowChange.toByteString());
                return entry;
            }
        } catch (Exception e) {
            throw new ParseException("parse row data failed.", e);
        }
    }

    private EntryPosition createPosition(Header logHeader) {
        return new EntryPosition(logHeader.getLogFileName(), logHeader.getLogPosition() - logHeader.getEventLength(), // startPos
                logHeader.getWhen() * 1000L,
                logHeader.getServerId()); // 记录到秒
    }

    private boolean parseOneRow(RowData.Builder rowDataBuilder, RowsEvent event, RowsLogBuffer buffer, BitSet cols,
                                boolean isAfter, TableMetaData tableMetaData) throws UnsupportedEncodingException {
        int columnCnt = event.getTableMapEvent().getColumnCnt();
        ColumnInfo[] columnInfo = event.getTableMapEvent().getColumnInfo();
        // mysql8.0针对set @@global.binlog_row_metadata='FULL' 可以记录部分的metadata信息
        boolean existOptionalMetaData = event.getTableMapEvent().isExistOptionalMetaData();
        boolean tableError = false;
        // check table fileds count，只能处理加字段
        boolean existRDSNoPrimaryKey = false;
        if (tableMetaData != null && columnInfo.length > tableMetaData.getColumnMetaDataList().size()) {
            if (tableMetaCache.isOnRDS()) {
                // 特殊处理下RDS的场景
                List<ColumnMetaData> primaryKeys = tableMetaData.getPrimaryKeyColumnMetaDataList();
                if (primaryKeys == null || primaryKeys.isEmpty()) {
                    if (columnInfo.length == tableMetaData.getColumnMetaDataList().size() + 1
                            && columnInfo[columnInfo.length - 1].type == BinLogEvent.MYSQL_TYPE_LONGLONG) {
                        existRDSNoPrimaryKey = true;
                    }
                }
            }

            EntryPosition position = createPosition(event.getHeader());
            if (!existRDSNoPrimaryKey) {
                // online ddl增加字段操作步骤：
                // 1. 新增一张临时表，将需要做ddl表的数据全量导入
                // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表
                // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作
                // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作
                // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作
                tableMetaData = getTableMeta(event.getTableMapEvent().getDbName(), event.getTableMapEvent().getTableName(), false, position);// 强制重新获取一次
                if (tableMetaData == null) {
                    tableError = true;
                    if (!filterTableError) {
                        throw new ParseException("not found [" + event.getTableMapEvent().getDbName() + "."
                                + event.getTableMapEvent().getTableName() + "] in db , pls check!");
                    }
                }

                // 在做一次判断
                if (tableMetaData != null && columnInfo.length > tableMetaData.getColumnMetaDataList().size()) {
                    tableError = true;
                    if (!filterTableError) {
                        throw new ParseException("column size is not match for table:" + tableMetaData.getFullName()
                                + "," + columnInfo.length + " vs " + tableMetaData.getColumnMetaDataList().size());
                    }
                }
                // } else {
                // logger.warn("[" + event.getTableName().getDbName() + "." +
                // event.getTableName().getTableName()
                // + "] is no primary key , skip alibaba_rds_row_id column");
            }
        }

        for (int i = 0; i < columnCnt; i++) {
            ColumnInfo info = columnInfo[i];
            // mysql 5.6开始支持nolob/mininal类型,并不一定记录所有的列,需要进行判断
            if (!cols.get(i)) {
                continue;
            }

            if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == BinLogEvent.MYSQL_TYPE_LONGLONG) {
                // 不解析最后一列
                String rdsRowIdColumnName = "#alibaba_rds_row_id#";
                buffer.nextValue(rdsRowIdColumnName, i, info.type, info.meta, false);
                Column.Builder columnBuilder = Column.newBuilder();
                columnBuilder.setName(rdsRowIdColumnName);
                columnBuilder.setIsKey(true);
                columnBuilder.setMysqlType("bigint");
                columnBuilder.setIndex(i);
                columnBuilder.setIsNull(false);
                Serializable value = buffer.getValue();
                columnBuilder.setValue(value.toString());
                columnBuilder.setSqlType(Types.BIGINT);
                columnBuilder.setUpdated(false);

                if (isAfter) {
                    rowDataBuilder.addAfterColumns(columnBuilder.build());
                } else {
                    rowDataBuilder.addBeforeColumns(columnBuilder.build());
                }
                continue;
            }

            ColumnMetaData columnMetaData = null;
            if (tableMetaData != null && !tableError) {
                // 处理file meta
                columnMetaData = tableMetaData.getColumnMetaDataList().get(i);
            }

            if (columnMetaData != null && existOptionalMetaData && tableMetaCache.isOnTSDB()) {
                // check column info
                boolean check = StringUtils.equalsIgnoreCase(columnMetaData.getColumnName(), info.name);
                check &= (columnMetaData.isUnsigned() == info.unsigned);
                check &= (columnMetaData.isNullable() == info.nullable);

                if (!check) {
                    throw new ParseException("MySQL8.0 unmatch column metadata & pls submit issue , table : "
                            + tableMetaData.getFullName() + ", db columnMetaData : "
                            + columnMetaData.toString() + " , binlog columnMetaData : " + info.toString()
                            + " , on : " + event.getHeader().getLogFileName() + ":"
                            + (event.getHeader().getLogPosition() - event.getHeader().getEventLength()));
                }
            }

            Column.Builder columnBuilder = Column.newBuilder();
            if (columnMetaData != null) {
                columnBuilder.setName(columnMetaData.getColumnName());
                columnBuilder.setIsKey(columnMetaData.isKey());
                // 增加mysql type类型,issue 73
                columnBuilder.setMysqlType(columnMetaData.getColumnType());
            } else if (existOptionalMetaData) {
                columnBuilder.setName(info.name);
                columnBuilder.setIsKey(info.pk);
                // mysql8.0里没有mysql type类型
                // columnBuilder.setMysqlType(columnMetaData.getColumnType());
            }
            columnBuilder.setIndex(i);
            columnBuilder.setIsNull(false);

            // fixed issue
            // https://github.com/alibaba/canal/issues/66，特殊处理binary/varbinary，不能做编码处理
            boolean isBinary = false;
            boolean isSingleBit = false;
            if (columnMetaData != null) {
                if (StringUtils.containsIgnoreCase(columnMetaData.getColumnType(), "VARBINARY")) {
                    isBinary = true;
                } else if (StringUtils.containsIgnoreCase(columnMetaData.getColumnType(), "BINARY")) {
                    isBinary = true;
                } else if (StringUtils.containsIgnoreCase(columnMetaData.getColumnType(), "TINYINT(1)")) {
                    isSingleBit = true;
                }
            }

            buffer.nextValue(columnBuilder.getName(), i, info.type, info.meta, isBinary);
            int javaType = buffer.getJavaType();
            if (buffer.isNull()) {
                columnBuilder.setIsNull(true);
            } else {
                final Serializable value = buffer.getValue();
                // 处理各种类型
                switch (javaType) {
                    case Types.INTEGER:
                    case Types.TINYINT:
                    case Types.SMALLINT:
                    case Types.BIGINT:
                        // 处理unsigned类型
                        Number number = (Number) value;
                        boolean isUnsigned = (columnMetaData != null ? columnMetaData.isUnsigned() : (existOptionalMetaData ? info.unsigned : false));
                        if (isUnsigned && number.longValue() < 0) {
                            switch (buffer.getLength()) {
                                case 1: /* MYSQL_TYPE_TINY */
                                    columnBuilder.setValue(String.valueOf(Integer.valueOf(TINYINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.SMALLINT; // 往上加一个量级
                                    break;

                                case 2: /* MYSQL_TYPE_SHORT */
                                    columnBuilder.setValue(String.valueOf(Integer.valueOf(SMALLINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.INTEGER; // 往上加一个量级
                                    break;

                                case 3: /* MYSQL_TYPE_INT24 */
                                    columnBuilder.setValue(String.valueOf(Integer.valueOf(MEDIUMINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.INTEGER; // 往上加一个量级
                                    break;

                                case 4: /* MYSQL_TYPE_LONG */
                                    columnBuilder.setValue(String.valueOf(Long.valueOf(INTEGER_MAX_VALUE
                                            + number.longValue())));
                                    javaType = Types.BIGINT; // 往上加一个量级
                                    break;

                                case 8: /* MYSQL_TYPE_LONGLONG */
                                    columnBuilder.setValue(BIGINT_MAX_VALUE.add(BigInteger.valueOf(number.longValue()))
                                            .toString());
                                    javaType = Types.DECIMAL; // 往上加一个量级，避免执行出错
                                    break;
                            }
                        } else {
                            // 对象为number类型，直接valueof即可
                            columnBuilder.setValue(String.valueOf(value));
                        }

                        if (isSingleBit && javaType == Types.TINYINT) {
                            javaType = Types.BIT;
                        }
                        break;
                    case Types.REAL: // float
                    case Types.DOUBLE: // double
                        // 对象为number类型，直接valueof即可
                        columnBuilder.setValue(String.valueOf(value));
                        break;
                    case Types.BIT:// bit
                        // 对象为number类型
                        columnBuilder.setValue(String.valueOf(value));
                        break;
                    case Types.DECIMAL:
                        columnBuilder.setValue(((BigDecimal) value).toPlainString());
                        break;
                    case Types.TIMESTAMP:
                        // 修复时间边界值
                        // String v = value.toString();
                        // v = v.substring(0, v.length() - 2);
                        // columnBuilder.setValue(v);
                        // break;
                    case Types.TIME:
                    case Types.DATE:
                        // 需要处理year
                        columnBuilder.setValue(value.toString());
                        break;
                    case Types.BINARY:
                    case Types.VARBINARY:
                    case Types.LONGVARBINARY:
                        // fixed text encoding
                        // https://github.com/AlibabaTech/canal/issues/18
                        // mysql binlog中blob/text都处理为blob类型，需要反查table
                        // meta，按编码解析text
                        if (columnMetaData != null && isText(columnMetaData.getColumnType())) {
                            columnBuilder.setValue(new String((byte[]) value, charset));
                            javaType = Types.CLOB;
                        } else {
                            // byte数组，直接使用iso-8859-1保留对应编码，浪费内存
                            columnBuilder.setValue(new String((byte[]) value, ISO_8859_1));
                            // columnBuilder.setValueBytes(ByteString.copyFrom((byte[])
                            // value));
                            javaType = Types.BLOB;
                        }
                        break;
                    case Types.CHAR:
                    case Types.VARCHAR:
                        columnBuilder.setValue(value.toString());
                        break;
                    default:
                        columnBuilder.setValue(value.toString());
                }
            }

            columnBuilder.setSqlType(javaType);
            // 设置是否update的标记位
            columnBuilder.setUpdated(isAfter
                    && isUpdate(rowDataBuilder.getBeforeColumnsList(),
                    columnBuilder.getIsNull() ? null : columnBuilder.getValue(),
                    i));
            if (isAfter) {
                rowDataBuilder.addAfterColumns(columnBuilder.build());
            } else {
                rowDataBuilder.addBeforeColumns(columnBuilder.build());
            }
        }

        return tableError;

    }

    private Entry buildQueryEntry(String queryString, Header logHeader, String tableName) {
        drds.propagate.protocol.CanalEntry.Header header = createHeader(logHeader, "", tableName, EventType.QUERY);
        RowChange.Builder rowChangeBuider = RowChange.newBuilder();
        rowChangeBuider.setSql(queryString);
        rowChangeBuider.setEventType(EventType.QUERY);
        return createEntry(header, EntryType.row_data, rowChangeBuider.build().toByteString());
    }

    private Entry buildQueryEntry(Header logHeader, String queryString) {
        drds.propagate.protocol.CanalEntry.Header header = createHeader(logHeader, "", "", EventType.QUERY);
        RowChange.Builder rowChangeBuider = RowChange.newBuilder();
        rowChangeBuider.setSql(queryString);
        rowChangeBuider.setEventType(EventType.QUERY);
        return createEntry(header, EntryType.row_data, rowChangeBuider.build().toByteString());
    }

    private drds.propagate.protocol.CanalEntry.Header createHeader(Header logHeader, String schemaName, String tableName, EventType eventType) {
        return createHeader(logHeader, schemaName, tableName, eventType, -1);
    }

    private drds.propagate.protocol.CanalEntry.Header createHeader(Header logHeader, String schemaName, String tableName, EventType eventType,
                                                                   Integer rowsCount) {
        // header会做信息冗余,方便以后做检索或者过滤
        drds.propagate.protocol.CanalEntry.Header.Builder headerBuilder = drds.propagate.protocol.CanalEntry.Header.newBuilder();
        headerBuilder.setVersion(version);
        headerBuilder.setLogfileName(logHeader.getLogFileName());
        // 记录的是该binlog的start offest
        headerBuilder.setLogfileOffset(logHeader.getLogPosition() - logHeader.getEventLength());
        headerBuilder.setServerId(logHeader.getServerId());
        headerBuilder.setServerenCode(UTF_8);// 经过java输出后所有的编码为unicode
        headerBuilder.setExecuteTime(logHeader.getWhen() * 1000L);
        headerBuilder.setSourceType(Type.MYSQL);
        if (eventType != null) {
            headerBuilder.setEventType(eventType);
        }
        if (schemaName != null) {
            headerBuilder.setSchemaName(schemaName);
        }
        if (tableName != null) {
            headerBuilder.setTableName(tableName);
        }
        headerBuilder.setEventLength(logHeader.getEventLength());
        // enable gtid position
        if (StringUtils.isNotEmpty(logHeader.getGtidSetStr())) {
            headerBuilder.setGtid(logHeader.getGtidSetStr());
        }
        // add current gtid
        if (StringUtils.isNotEmpty(logHeader.getCurrentGtid())) {
            Pair pair = createSpecialPair("curtGtid", logHeader.getCurrentGtid());
            headerBuilder.addProps(pair);
        }
        // add current gtid sequence no
        if (StringUtils.isNotEmpty(logHeader.getCurrentGtidSn())) {
            Pair pair = createSpecialPair("curtGtidSn", logHeader.getCurrentGtidSn());
            headerBuilder.addProps(pair);
        }

        // add current gtid last committed
        if (StringUtils.isNotEmpty(logHeader.getCurrentGtidLastCommit())) {
            Pair pair = createSpecialPair("curtGtidLct", logHeader.getCurrentGtidLastCommit());
            headerBuilder.addProps(pair);
        }

        // add rowsCount suppport
        if (rowsCount > 0) {
            Pair pair = createSpecialPair("rowsCount", String.valueOf(rowsCount));
            headerBuilder.addProps(pair);
        }
        return headerBuilder.build();
    }

    private boolean isUpdate(List<Column> bfColumns, String newValue, int index) {
        if (bfColumns == null) {
            throw new ParseException("ERROR ## the bfColumns is null");
        }

        if (index < 0) {
            return false;
        }

        for (Column column : bfColumns) {
            if (column.getIndex() == index) {// 比较before / after的column binlog_position
                if (column.getIsNull() && newValue == null) {
                    // 如果全是null
                    return false;
                } else if (newValue != null && (!column.getIsNull() && column.getValue().equals(newValue))) {
                    // fixed issue #135, old column is Null
                    // 如果不为null，并且相等
                    return false;
                }
            }
        }

        // 比如nolob/minial模式下,可能找不到before记录,认为是有变化
        return true;
    }

    private TableMetaData getTableMeta(String dbName, String tbName, boolean useCache, EntryPosition position) {
        try {
            return tableMetaCache.getTableMeta(dbName, tbName, useCache, position);
        } catch (Throwable e) {
            String message = ExceptionUtils.getRootCauseMessage(e);
            if (filterTableError) {
                if (StringUtils.contains(message, "errorNumber=1146") && StringUtils.contains(message, "doesn't exist")) {
                    return null;
                } else if (StringUtils.contains(message, "errorNumber=1142")
                        && StringUtils.contains(message, "command denied")) {
                    return null;
                }
            }

            throw new ParseException(e);
        }
    }

    private boolean isText(String columnType) {
        return "LONGTEXT".equalsIgnoreCase(columnType) || "MEDIUMTEXT".equalsIgnoreCase(columnType)
                || "TEXT".equalsIgnoreCase(columnType) || "TINYTEXT".equalsIgnoreCase(columnType);
    }

    private boolean isAliSQLHeartBeat(String schema, String table) {
        return "test".equalsIgnoreCase(schema) && "heartbeat".equalsIgnoreCase(table);
    }

    private boolean isRDSHeartBeat(String schema, String table) {
        return "mysql".equalsIgnoreCase(schema) && "ha_health_check".equalsIgnoreCase(table);
    }

    public void setCharset(Charset charset) {
        this.charset = charset;
    }

    public void setNameFilter(AviaterRegexFilter nameFilter) {
        this.nameFilter = nameFilter;
        logger.warn("--> init table filter : " + nameFilter.toString());
    }

    public void setNameBlackFilter(AviaterRegexFilter nameBlackFilter) {
        this.nameBlackFilter = nameBlackFilter;
        logger.warn("--> init table black filter : " + nameBlackFilter.toString());
    }

    public void setTableMetaCache(TableMetaCache tableMetaCache) {
        this.tableMetaCache = tableMetaCache;
    }

    public void setFilterQueryDcl(boolean filterQueryDcl) {
        this.filterQueryDcl = filterQueryDcl;
    }

    public void setFilterQueryDml(boolean filterQueryDml) {
        this.filterQueryDml = filterQueryDml;
    }

    public void setFilterQueryDdl(boolean filterQueryDdl) {
        this.filterQueryDdl = filterQueryDdl;
    }

    public void setFilterTableError(boolean filterTableError) {
        this.filterTableError = filterTableError;
    }

    public void setFilterRows(boolean filterRows) {
        this.filterRows = filterRows;
    }

    public void setUseDruidDdlFilter(boolean useDruidDdlFilter) {
        this.useDruidDdlFilter = useDruidDdlFilter;
    }
}
